#!/usr/bin/python3
# -*- coding: utf-8; tab-width: 4; indent-tabs-mode: t -*-

import os
import sys
import time
import dbus
import json
import stat
import prctl
import ctypes
import struct
import shutil
import psutil
import socket
import aiohttp
import hashlib
import asyncio
import logging
import importlib
import traceback
import subprocess
import collections
from typing import Set
from typing import Final
from multidict import istr
from gi.repository import GLib
from dbus.mainloop.glib import DBusGMainLoop


class McUtil:

    @staticmethod
    def asyncProcTerminate(proc):
        try:
            proc.terminate()
        except ProcessLookupError:
            # it seems asyncio has this exception, subprocess.Popen never raises this
            pass

    @staticmethod
    async def asyncSleepForever():
        # it is astonishing that there's no build-in function for this, sucks
        while True:
            await asyncio.sleep(3600)

    @classmethod
    def asyncSafeTaskCreate(cls, loop, coro, taskName=None, errPrefix=None, bNoCancel=False, bTerminateOnError=False, bTerminateOnEnd=False):
        if bTerminateOnEnd:
            assert bTerminateOnError
        ret = loop.create_task(cls._safeTaskMain(coro, errPrefix), name=taskName)
        ret._mcUtilNoCancel = bNoCancel
        ret._mcUtilTerminateOnError = bTerminateOnError
        ret._mcUtilTerminateOnEnd = bTerminateOnEnd
        return ret

    @classmethod
    def asyncSafeTaskSetNoCancel(cls, task, bNoCancel):
        task._mcUtilNoCancel = bNoCancel

    @classmethod
    def asyncSafeTaskSetTerminateOnError(cls, task, bTerminateOnError):
        if not bTerminateOnError:
            assert not task._mcUtilTerminateOnEnd
        task._mcUtilTerminateOnError = bTerminateOnError

    @classmethod
    def asyncSafeTaskSetTerminateOnEnd(cls, task, bTerminateOnEnd):
        if bTerminateOnEnd:
            assert task._mcUtilTerminateOnError
        task._mcUtilTerminateOnEnd = bTerminateOnEnd

    @staticmethod
    async def _safeTaskMain(coro, errPrefix):
        try:
            try:
                await coro
                if asyncio.current_task()._mcUtilTerminateOnEnd:
                    print("Task %s ends unexpectedly." % (asyncio.current_task().name))
                    # terminate program by stopping the event loop
                    loop = asyncio.get_running_loop()
                    loop.call_soon(loop.stop)
            except asyncio.CancelledError:
                assert not asyncio.current_task()._mcUtilNoCancel
                raise
            finally:
                # no task should be cancelled twice
                assert asyncio.current_task().cancelling() <= 1
        except asyncio.CancelledError:
            # can-be-cancelled task must be awaited by others, so we can safely throw this exception out
            raise
        except BaseException as e:
            if asyncio.current_task()._mcUtilTerminateOnError:
                # simulate standard exception and stack traceback print out
                print("Traceback (most recent call last):")
                traceback.print_tb(e.__traceback__)
                print("%s: %s" % (e.__class__.__name__, f"{errPrefix}, {e}" if errPrefix is not None else f"{e}"))
                # terminate program by stopping the event loop
                loop = asyncio.get_running_loop()
                loop.call_soon(loop.stop)
            else:
                raise

    @staticmethod
    def getExistFile(filename):
        assert os.path.exists(filename)
        return filename

    @staticmethod
    def findSocketFileInDir(dirpath):
        for fn in os.listdir(dirpath):
            fullfn = os.path.join(dirpath, fn)
            if stat.S_ISSOCK(os.stat(fullfn).st_mode):
                return fullfn
        assert False

    @staticmethod
    def rotateLogCmdList(baseFilename, maxBytes, backupCount):
        return ["/usr/sbin/rotatelogs", "-f", "-n", str(backupCount), baseFilename, str(maxBytes) + "B"]

    @staticmethod
    def createRotateLogProcess(baseFilename, maxBytes, backupCount, outf):
        # rotate firstly
        for i in range(backupCount - 1, 0, -1):
            sfn = "%s.%d" % (baseFilename, i)
            dfn = "%s.%d" % (baseFilename, i + 1)
            if os.path.exists(dfn):
                os.remove(dfn)
            if os.path.exists(sfn):
                os.rename(sfn, dfn)
        dfn = baseFilename + ".1"
        if os.path.exists(dfn):
            os.remove(dfn)
        if os.path.exists(baseFilename):
            os.rename(baseFilename, dfn)

        # create process
        return subprocess.Popen(["/usr/sbin/rotatelogs", "-f", "-n", str(backupCount), baseFilename, str(maxBytes) + "B"], stdin=outf)

    @staticmethod
    def loadPythonFile(filename):
        # returns (mod-name,mod-object)
        modname = os.path.basename(filename).replace(".py", "").replace("-", "_")
        spec = importlib.util.spec_from_file(modname, filename)
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        return (modname, mod)

    @staticmethod
    def stdTmFmt():
        return "%Y-%m-%d %H:%M:%S"

    @staticmethod
    def sqlInsertStatement(tableName, valueDict):
        return "INSERT INTO " + tableName + " (" + ",".join(valueDict.keys()) + ") VALUES ('" + "','".join(valueDict.values()) + "');"

    @staticmethod
    def mysqlPrivJson(password=None):
        if password is None:
            return '{"access":0}'
        else:
            # password_last_changed must be >0, or else mysql would force the user change his/her password
            return '{"access":0,"plugin":"mysql_native_password","authentication_string":"%s","password_last_changed":1}' % (McUtil.mysqlPasswordHash(password))

    @staticmethod
    def mysqlPasswordHash(password):
        r = hashlib.sha1(password.encode("iso8859-1")).digest()
        r = hashlib.sha1(r).hexdigest()
        return '*' + r

    @staticmethod
    def readFile(filename, defaultContent=None):
        if not os.path.exists(filename) and defaultContent is not None:
            return defaultContent
        with open(filename) as f:
            return f.read()

    @staticmethod
    def rreplace(s, sub, dst, count):
        # https://stackoverflow.com/questions/9943504/right-to-left-string-replace-in-python
        return dst.join(s.rsplit(sub, count))

    @staticmethod
    def cmdCall(cmd, *kargs):
        # call command to execute backstage job
        #
        # scenario 1, process group receives SIGTERM, SIGINT and SIGHUP:
        #   * callee must auto-terminate, and cause no side-effect
        #   * caller must be terminated by signal, not by detecting child-process failure
        # scenario 2, caller receives SIGTERM, SIGINT, SIGHUP:
        #   * caller is terminated by signal, and NOT notify callee
        #   * callee must auto-terminate, and cause no side-effect, after caller is terminated
        # scenario 3, callee receives SIGTERM, SIGINT, SIGHUP:
        #   * caller detects child-process failure and do appopriate treatment

        ret = subprocess.run([cmd] + list(kargs),
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                             universal_newlines=True)
        if ret.returncode > 128:
            # for scenario 1, caller's signal handler has the oppotunity to get executed during sleep
            time.sleep(1.0)
        if ret.returncode != 0:
            print(ret.stdout)
            ret.check_returncode()
        return ret.stdout.rstrip()

    @staticmethod
    def cmdCallWithInput(cmd, inStr, *kargs):
        ret = subprocess.run([cmd] + list(kargs),
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                             input=inStr, universal_newlines=True)
        if ret.returncode > 128:
            time.sleep(1.0)
        if ret.returncode != 0:
            print(ret.stdout)
            ret.check_returncode()
        return ret.stdout.rstrip()

    @staticmethod
    def preparePersistDir(dirname, uid, gid, mode):
        if not os.path.exists(dirname):
            os.makedirs(dirname, mode)
            if os.getuid() != uid or os.getpid() != gid:
                os.chown(dirname, uid, gid)
        else:
            st = os.stat(dirname)
            if stat.S_IMODE(st.st_mode) != mode:
                os.chmod(dirname, mode)
            if st.st_uid != uid or st.st_gid != gid:
                os.chown(dirname, uid, gid)
                for root, dirs, files in os.walk(dirname):
                    for d in dirs:
                        os.lchown(os.path.join(root, d), uid, gid)
                    for f in files:
                        os.lchown(os.path.join(root, f), uid, gid)

    @staticmethod
    def prepareTransientDir(dirname, uid, gid, mode):
        McUtil.forceDelete(dirname)
        os.makedirs(dirname, mode)
        if os.getuid() != uid or os.getpid() != gid:
            os.chown(dirname, uid, gid)

    @staticmethod
    def getUnixDomainSocketPeerInfo(sock):
        # returns (pid, uid, gid)
        pattern = "=iii"
        length = struct.calcsize(pattern)
        ret = sock.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, length)
        return struct.unpack(pattern, ret)

    @staticmethod
    def forceDelete(filename):
        if os.path.islink(filename):
            os.remove(filename)
        elif os.path.isfile(filename):
            os.remove(filename)
        elif os.path.isdir(filename):
            shutil.rmtree(filename)

    @staticmethod
    def mkDirAndClear(dirname):
        McUtil.forceDelete(dirname)
        os.mkdir(dirname)

    @staticmethod
    def getFreeSocketPort(portType):
        if portType == "tcp":
            stlist = [socket.SOCK_STREAM]
        elif portType == "udp":
            stlist = [socket.SOCK_DGRAM]
        elif portType == "tcp+udp":
            stlist = [socket.SOCK_STREAM, socket.SOCK_DGRAM]
        else:
            assert False

        for port in range(10000, 65536):
            bFound = True
            for sType in stlist:
                s = socket.socket(socket.AF_INET, sType)
                try:
                    s.bind((('', port)))
                except socket.error:
                    bFound = False
                finally:
                    s.close()
            if bFound:
                return port

        raise Exception("no valid port")

    @staticmethod
    def waitSocketPortForProc(portType, ip, port, proc, timeout=10):
        assert portType in ["tcp", "udp"]
        for i in range(0, timeout*10):
            if proc.poll() is not None:
                raise Exception("process terminated")
            time.sleep(0.1)
            for c in psutil.net_connections(kind=portType):
                # if c.pid == proc.pid and c.status == "LISTEN":        # FIXME: for apache, c.pid is wrong, result of netstat is right
                if c.status == "LISTEN":
                    if c.family == socket.AF_INET and c.laddr[0] == ip and c.laddr[1] == port:
                        return
                    if ip == "0.0.0.0" and c.family == socket.AF_INET6 and c.laddr[0] == "::" and c.laddr[1] == port:
                        return
        raise Exception("timeout")

    @staticmethod
    def touchFile(filename):
        if not os.path.exists(filename):
            with open(filename, 'w'):
                pass

    @staticmethod
    def getSyscallNumber(syscallName):
        # syscallName example: "SYS_prctl"
        out = McUtil.cmdCallWithInput("/usr/bin/gcc", "#include <sys/syscall.h>\n%s" % (syscallName), "-E", "-")
        syscall_number = out.split("\n")[-1]
        try:
            syscall_number = int(syscall_number)
        except ValueError:
            raise Exception("failed to get syscall number for %s" % (syscallName))
        if not 0 <= syscall_number <= 999:
            raise Exception("failed to get syscall number for %s" % (syscallName))
        return syscall_number


class StdoutRedirector:

    def __init__(self, filename):
        self.terminal = sys.stdout
        self.log = open(filename, "a")

    def write(self, message):
        self.terminal.write(message)
        self.log.write(message)
        self.log.flush()

    def flush(self):
        self.terminal.flush()
        self.log.flush()

    def isatty(self):
        return self.terminal.isatty()


class DynObject:
    # an object that can contain abitrary dynamically created properties and methods
    pass


class DropPriviledge:

    def __init__(self, uid, gid, caps=[]):
        assert os.getuid() == 0
        assert os.getgid() == 0

        self._oldInheritable = None
        self._oldAmbient = None
        self._oldKeepCaps = None
        self._oldNoSetuidFixup = None

        if len(caps) > 0:
            assert caps == [prctl.CAP_NET_BIND_SERVICE]                     # FIXME

            # self._oldInheritable =                                        # FIXME
            # self._oldAmbient =                                            # FIXME
            self._oldKeepCaps = prctl.securebits.keep_caps
            self._oldNoSetuidFixup = prctl.securebits.no_setuid_fixup
            prctl.cap_inheritable.net_bind_service = True                   # FIXME, prctl.cap_inheritable.limit() has no effect
            self._capAmbientRaise(caps)                                     # FIXME
            prctl.securebits.keep_caps = True
            prctl.securebits.no_setuid_fixup = True

        os.setresgid(gid, gid, 0)       # must change gid first
        os.setresuid(uid, uid, 0)

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        os.setuid(0)
        os.setgid(0)

        if self._oldNoSetuidFixup is not None:
            prctl.securebits.no_setuid_fixup = self._oldNoSetuidFixup
        if self._oldKeepCaps is not None:
            prctl.securebits.keep_caps = self._oldKeepCaps
        if self._oldAmbient is not None:
            assert False            # FIXME
        if self._oldInheritable is not None:
            assert False            # FIXME

    def _capAmbientRaise(self, caps):
        # this function calls SYS_prctl directly, because ambient set is not supported by prctl module yet

        _prctl = ctypes.CDLL(None).syscall
        _prctl.restype = ctypes.c_int
        _prctl.argtypes = ctypes.c_int, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong

        _PR_CAP_AMBIENT = 47            # from <linux/prctl.h>
        _PR_CAP_AMBIENT_RAISE = 2       # from <linux/prctl.h>

        for cap in caps:
            _prctl(McUtil.getSyscallNumber("SYS_prctl"), _PR_CAP_AMBIENT, _PR_CAP_AMBIENT_RAISE, cap, 0, 0)


class UnixDomainSocketApiServer:

    def __init__(self, serverFile, clientAppearFunc, clientDisappearFunc, notifyFunc):
        # Parameter clientAppearFunc is called after client appears.
        # Parameter clientDisappearFunc is called after we find client disappears and before we destroy the client object.
        # Parameter clientDisappearFunc can be None.

        assert serverFile is not None
        assert clientAppearFunc is not None and notifyFunc is not None

        self.clientAppearFunc = clientAppearFunc
        self.clientDisappearFunc = clientDisappearFunc
        self.notifyFunc = notifyFunc

        self.serverSock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.serverSock.bind(serverFile)
        self.serverSock.listen(5)
        self.serverSourceId = GLib.io_add_watch(self.serverSock, GLib.IO_IN, self.onServerAccept)

        self.clientInfoDict = dict()

    def dispose(self):
        for sock, obj in self.clientInfoDict.items():
            GLib.source_remove(obj.inWatch)
            sock.close()
        GLib.source_remove(self.serverSourceId)
        self.serverSock.close()

    def onServerAccept(self, source, cb_condition):
        # event callback, no exception is allowed

        new_sock, addr = source.accept()

        try:
            data = self.clientAppearFunc(new_sock)
        except BaseException:
            # absorb exception raised by upper layer function
            traceback.print_exc()
            new_sock.close()
            return True

        obj = DynObject()
        obj.inWatch = GLib.io_add_watch(new_sock, GLib.IO_IN | GLib.IO_PRI | GLib.IO_ERR | GLib.IO_HUP | GLib.IO_NVAL, self.onRecv)
        obj.recvBuf = b''
        obj.clientData = data
        self.clientInfoDict[new_sock] = obj
        return True

    def onRecv(self, source, cb_condition):
        bCloseSocket = False
        try:
            obj = self.clientInfoDict[source]

            # receive and parse
            obj.recvBuf += source.recv(4096)
            while True:
                i = obj.recvBuf.find(b'\n')
                if i < 0:
                    break
                jsonObj = json.loads(obj.recvBuf[:i].decode("utf-8"))
                obj.recvBuf = obj.recvBuf[i + 1:]
                try:
                    self.notifyFunc(obj.clientData, jsonObj)
                except BaseException:
                    # absorb exception raised by upper layer function, FIXME
                    print("upper layer exception")
                    traceback.print_exc()
                    raise

            # remote closed
            if (cb_condition & GLib.IO_HUP):
                bCloseSocket = True
                if len(obj.recvBuf) > 0:
                    raise Exception("remote close")
        except BaseException:
            print("excp IO_IN, %d" % (cb_condition & GLib.IO_IN))               # FIXME
            print("excp IO_PRI, %d" % (cb_condition & GLib.IO_PRI))
            print("excp IO_ERR, %d" % (cb_condition & GLib.IO_ERR))
            print("excp IO_HUP, %d" % (cb_condition & GLib.IO_HUP))
            print("excp IO_NVAL, %d" % (cb_condition & GLib.IO_NVAL))
            traceback.print_exc()
        finally:
            if bCloseSocket:
                if self.clientDisappearFunc is not None:
                    self.clientDisappearFunc(self.clientInfoDict[source].clientData)
                del self.clientInfoDict[source]
                source.close()
                return False
            else:
                return True


class AvahiServiceRegister:

    """
    Exampe:
        obj = AvahiServiceRegister()
        obj.add_service(socket.gethostname(), "_http", 80)
        obj.start()
        obj.stop()
    """

    def __init__(self):
        self.retryInterval = 30
        self.serviceList = []

        self._server = None
        self._retryCreateServerTimer = None
        self._entryGroup = None
        self._retryRegisterServiceTimer = None
        self._ownerChangeHandler = None

    def add_service(self, service_name, service_type, port):
        assert isinstance(service_name, str)
        assert service_type.endswith("._tcp") or service_type.endswith("._udp")
        assert isinstance(port, int)
        self.serviceList.append((service_name, service_type, port))

    def start(self):
        DBusGMainLoop(set_as_default=True)

        if dbus.SystemBus().name_has_owner("org.freedesktop.Avahi"):
            self._createServer()
        self._ownerChangeHandler = dbus.SystemBus().add_signal_receiver(self.onNameOwnerChanged, "NameOwnerChanged", None, None)

    def stop(self):
        if self._ownerChangeHandler is not None:
            dbus.SystemBus().remove_signal_receiver(self._ownerChangeHandler)
            self._ownerChangeHandler = None
        self._unregisterService()
        self._releaseServer()

    def onNameOwnerChanged(self, name, old, new):
        if name == "org.freedesktop.Avahi":
            if new != "" and old == "":
                if self._server is None:
                    self._createServer()
                else:
                    # this may happen on some rare case
                    pass
            elif new == "" and old != "":
                self._unregisterService()
                self._releaseServer()
            else:
                assert False

    def _createServer(self):
        assert self._server is None and self._retryCreateServerTimer is None
        assert self._entryGroup is None
        try:
            self._server = dbus.Interface(dbus.SystemBus().get_object("org.freedesktop.Avahi", "/"), "org.freedesktop.Avahi.Server")
            if self._server.GetState() == 2:    # avahi.SERVER_RUNNING
                self._registerService()
            self._server.connect_to_signal("StateChanged", self.onSeverStateChanged)
        except Exception:
            logging.error("Avahi create server failed, retry in %d seconds" % (self.retryInterval), exc_info=True)
            self._releaseServer()
            self._retryCreateServer()

    def _releaseServer(self):
        assert self._entryGroup is None
        if self._retryCreateServerTimer is not None:
            GLib.source_remove(self._retryCreateServerTimer)
            self._retryCreateServerTimer = None
        self._server = None

    def onSeverStateChanged(self, state, error):
        if state == 2:      # avahi.SERVER_RUNNING
            self._unregisterService()
            self._registerService()
        else:
            self._unregisterService()

    def _registerService(self):
        assert self._entryGroup is None and self._retryRegisterServiceTimer is None
        try:
            self._entryGroup = dbus.Interface(dbus.SystemBus().get_object("org.freedesktop.Avahi", self._server.EntryGroupNew()),
                                              "org.freedesktop.Avahi.EntryGroup")
            for serviceName, serviceType, port in self.serviceList:
                self._entryGroup.AddService(-1,                 # interface = avahi.IF_UNSPEC
                                            0,                  # protocol = avahi.PROTO_UNSPEC
                                            dbus.UInt32(0),     # flags
                                            serviceName,        # name
                                            serviceType,        # type
                                            "",                 # domain
                                            "",                 # host
                                            dbus.UInt16(port),  # port
                                            "")                 # txt
            self._entryGroup.Commit()
            self._entryGroup.connect_to_signal("StateChanged", self.onEntryGroupStateChanged)
        except Exception:
            logging.error("Avahi register service failed, retry in %d seconds" % (self.retryInterval), exc_info=True)
            self._unregisterService()
            self._retryRegisterService()

    def _unregisterService(self):
        if self._retryRegisterServiceTimer is not None:
            GLib.source_remove(self._retryRegisterServiceTimer)
            self._retryRegisterServiceTimer = None
        if self._entryGroup is not None:
            try:
                if self._entryGroup.GetState() != 4:        # avahi.ENTRY_GROUP_FAILURE
                    self._entryGroup.Reset()
                    self._entryGroup.Free()
                    # .Free() has mem leaks?
                    self._entryGroup._obj._bus = None
                    self._entryGroup._obj = None
            except dbus.exceptions.DBusException:
                pass
            finally:
                self._entryGroup = None

    def onEntryGroupStateChanged(self, state, error):
        if state in [0, 1, 2]:  # avahi.ENTRY_GROUP_UNCOMMITED, avahi.ENTRY_GROUP_REGISTERING, avahi.ENTRY_GROUP_ESTABLISHED
            pass
        elif state == 3:        # avahi.ENTRY_GROUP_COLLISION
            self._unregisterService()
            self._retryRegisterService()
        elif state == 4:        # avahi.ENTRY_GROUP_FAILURE
            assert False
        else:
            assert False

    def _retryCreateServer(self):
        assert self._retryCreateServerTimer is None
        self._retryCreateServerTimer = GLib.timeout_add_seconds(self.retryInterval, self.__timeoutCreateServer)

    def __timeoutCreateServer(self):
        self._retryCreateServerTimer = None
        self._createServer()                    # no exception in self._createServer()
        return False

    def _retryRegisterService(self):
        assert self._retryRegisterServiceTimer is None
        self._retryRegisterServiceTimer = GLib.timeout_add_seconds(self.retryInterval, self.__timeoutRegisterService)

    def __timeoutRegisterService(self):
        self._retryRegisterServiceTimer = None
        self._registerService()                 # no exception in self._registerService()
        return False


# ASGI is a more modernized standard than aiohttp, I believe it would support every common or uncommon component, such as CGI, FastCGI, etc.
# And, aiohttp says "In production, static content should be processed by web servers like nginx or apache.", but we need to serve static files.
# but starlette still lacks many things
StarletteHdrs = aiohttp.hdrs


# starlette does not support route decorations. why?
# TODO: implement route decorations, it should add some properties to the route method
class StarletteRouteDecorations:
    pass


# from https://github.com/accent-starlette/starlette-core (which is dead)
class Paginator:

    def __init__(self, object_list, per_page):
        self.object_list = object_list
        self.per_page = int(per_page)
        self._count = None

    def validate_number(self, number):
        """Validate the given 1-based page number."""

        try:
            number = int(number)
        except (TypeError, ValueError):
            raise PageNotAnInteger("That page number is not an integer")

        if number < 1:
            raise EmptyPage("That page number is less than 1")
        if number > self.num_pages and number != 1:
            raise EmptyPage("That page contains no results")

        return number

    def get_page(self, number):
        """
        Return a valid page, even if the page argument isn't a number or isn't
        in range.
        """

        try:
            number = self.validate_number(number)
        except PageNotAnInteger:
            number = 1
        except EmptyPage:
            number = self.num_pages
        return self.page(number)

    def page(self, number):
        """Return a Page object for the given 1-based page number."""

        number = self.validate_number(number)
        bottom = (number - 1) * self.per_page
        top = bottom + self.per_page
        return self._get_page(self.object_list[bottom:top], number, self)

    def _get_page(self, *args, **kwargs):
        """
        Return an instance of a single page.
        This hook can be used by subclasses to use an alternative to the
        standard :cls:`Page` object.
        """

        return Page(*args, **kwargs)

    @property
    def count(self):
        """Return the total number of objects, across all pages."""

        if self._count:
            return self._count

        c = getattr(self.object_list, "count", None)
        if callable(c) and not inspect.isbuiltin(c) and method_has_no_args(c):
            self._count = c()
        else:
            self._count = len(self.object_list)

        return self._count

    @property
    def num_pages(self):
        """Return the total number of pages."""

        hits = max(1, self.count)
        return ceil(hits / self.per_page)


class Page(collections.abc.Sequence):

    def __init__(self, object_list, number, paginator):
        self.object_list = object_list
        self.number = number
        self.paginator = paginator

    def __repr__(self):
        return f"<Page {self.number} of {self.paginator.num_pages}>"

    def __len__(self):
        return len(self.object_list)

    def __getitem__(self, index):
        if not isinstance(index, (int, slice)):
            raise TypeError
        # The object_list is converted to a list so that if it was a QuerySet
        # it won't be a database hit per __getitem__.
        if not isinstance(self.object_list, list):
            self.object_list = list(self.object_list)
        return self.object_list[index]

    @property
    def has_next(self):
        return self.number < self.paginator.num_pages

    @property
    def next_page_number(self):
        return self.paginator.validate_number(self.number + 1)

    @property
    def has_previous(self):
        return self.number > 1

    @property
    def previous_page_number(self):
        return self.paginator.validate_number(self.number - 1)

    @property
    def has_other_pages(self):
        return self.has_previous or self.has_next